package query;

import global.SortKey;
import heap.HeapFile;
import parser.AST_Select;
import relop.FileScan;
import relop.Iterator;
import relop.Predicate;
import relop.Projection;
import relop.Schema;
import relop.Selection;
import relop.SimpleJoin;

/**
 * Execution plan for selecting tuples.
 */
class Select implements Plan {
  public boolean isExplain;
  public boolean isDistinct;
  protected String[] columns;
  protected String[] tables;
  protected Predicate[][] predicates;
  protected Schema[] tableSchemas;
  protected Schema schema;
  protected SortKey[] orders;
  protected Iterator iter;
  protected Integer[] projectFields;
  
  protected FileScan[] tableScans;
  protected SimpleJoin[] joins;
  protected Projection projection;
  protected Selection[] selection;
  /**
   * Optimizes the plan, given the parsed query.
   * 
   * @throws QueryException if validation fails
   */
  public Select(AST_Select tree) throws QueryException {
    isExplain = tree.isExplain;
    isDistinct = tree.isDistinct;
    tables = tree.getTables();

    tableSchemas = new Schema[tables.length];
    // create the schema for the simple join table
    for (int i = 0; i < tables.length; i++) {
      String table = tables[i];
      tableSchemas[i] = QueryCheck.tableExists(table);
      if (null == tableSchemas[i]) throw new QueryException("Invalid Schema");
      if (i == 0)
    	  schema = tableSchemas[i];
      else
    	  schema = Schema.join(schema, tableSchemas[i]);
    }
    
    columns = tree.getColumns();
    // if * then length is zero. Fill all fields.
    if (columns.length == 0) {
    	columns = new String[schema.getCount()];
    	for (int i = 0; i < columns.length; i++) {
    		columns[i] = schema.fieldName(i);
    	}
    }
    
    projectFields = new Integer[columns.length];
    // check if the project columns exists
    for (int i = 0; i < columns.length; i++) {
      projectFields[i] = QueryCheck.columnExists(schema, columns[i]);
    }
    predicates = tree.getPredicates();

    // check if the predicates exists
    QueryCheck.predicates(schema, predicates);

    //TODO how to validate? do we have to check with the joined schema?
    orders = tree.getOrders();

    // Creating file scans for tables
    tableScans = new FileScan[tables.length];
    for (int i = 0; i < tables.length; i++) {
      tableScans[i] = new FileScan(tableSchemas[i], new HeapFile(tables[i]));
    }
    
   	iter = tableScans[0];
   	
    if (tables.length > 1) {
	    // Simple Join
	    joins = new SimpleJoin[tables.length-1];
	    for (int i = 0; i < tables.length - 1; i++) {
	      if (i == 0)
	        joins[i] = new SimpleJoin(tableScans[i], tableScans[i+1]);
	      else
	        joins[i] = new SimpleJoin(joins[i-1], tableScans[i+1]);
	    }
	    iter = joins[joins.length-1];
    }
    
    if (predicates.length > 0) {
	    selection = new Selection[predicates.length];
	    for (int i = 0; i < predicates.length; i++) {
	    	if (i == 0)
	    		selection[i] = new Selection(iter, predicates[i]);
	    	else
	    		selection[i] = new Selection(selection[i-1], predicates[i]);
	    }
	    iter = selection[selection.length-1];
    }
   
    projection = new Projection(iter, projectFields);
    
    iter = projection;
  } // public Select(AST_Select tree) throws QueryException

  /**
   * Executes the plan and prints applicable output.
   */
  public void execute() {
    if (isExplain) iter.explain(0);
    else {
      int numRows = iter.execute();
      // print the output message
      System.out.println(numRows + " rows affected.");
    }

  } // public void execute()

} // class Select implements Plan
